package com.speily.study.provider.rocketmq;

import com.speily.study.provider.entity.MessageTransaction;
import com.speily.study.provider.service.MessageTransactionService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @Auther: SPL
 * @Date: 2019-07-10 13:14
 * @Description: 实现TransactionListener事务监听器，主要是确定本地事物处理没问题
 */
@Slf4j
public class TransactionListenerImpl implements TransactionListener {

    /**
     * 说明：表t_message_transaction，用于记录本地事物执行状态，使用事物ID作为主键即可，回查事物时使用。
     */

    private final static int MAX_COUNT = 5;

    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

    @Autowired
    private MessageTransactionService messageTransactionService;


    /**
     * 发送prepare消息成功后回调该方法用于执行本地事务，该方法与业务方代码在一个事务中
     * @param msg 回传的消息，利用transactionId即可获取到该消息的唯一Id
     * @param o 调用send方法时传递的参数，当send时候若有额外的参数可以传递到send方法中，这里能获取到
     * @return 返回事务状态，COMMIT：提交  ROLLBACK：回滚  UNKNOW：回调
     */
    public LocalTransactionState executeLocalTransaction(Message msg, Object o) {


        // 记录本地消息事物提交状态，t_message_transaction添加一条记录（事物Id唯一），在事务回查时，如果存在记录，就认为是该消息需要提交。
        messageTransactionService.insertOne(MessageTransaction.builder()
                .topic(msg.getTopic())
                .id(msg.getTransactionId())
        );
        localTrans.put(msg.getTransactionId(), 0);
        log.info("处理prepare消息，消息事物ID：{}",msg.getTransactionId());

        System.out.println("执行 executeLocalTransaction 方法");
        return LocalTransactionState.UNKNOW;
    }

    /**
     * 检查本地事务<br>
     *     UNKNOW 状态：表示事务消息未确定，可能是业务方执行本地事务逻辑时间耗时过长或者网络原因等引起的，该状态会导致broker对事务消息进行回查，默认回查总次数是15次，第一次回查间隔时间是6s，后续每次间隔60s,
     *     ROLLBACK 状态，该状态表示该事务消息被回滚，因为本地事务逻辑执行失败导致
     *     COMMIT 状态，表示事务消息被提交，会被正确分发给消费者。
     * @param msg
     * @return
     */
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {

        // 查询本地消息事物提交状态，t_message_transaction（事物Id唯一）select count(1) from t_message_transaction where id = xxx
        // 如果存在，则返回COMMIT_MESSAGE，
        // 不存在，则记录查询次数，未超过次数，返回UNKNOW，超过次数，返回ROLLBACK_MESSAGE
        MessageTransaction mt = messageTransactionService.selectById(msg.getTransactionId());

        if (mt != null) {
            return LocalTransactionState.COMMIT_MESSAGE;//消息接收端提交事物（消费消息）
        }else {
            int num = localTrans.get(msg.getTransactionId());
            if(num < MAX_COUNT){
                return LocalTransactionState.UNKNOW;//未知重试
            }else {
                return LocalTransactionState.ROLLBACK_MESSAGE;//消息接收端放弃提交事物（消费消息）
            }
        }
    }
}
